package com.cloud.rocketmqtxconsumer.listener;

import com.alibaba.fastjson.JSONObject;
import com.cloud.rocketmqtxconsumer.entity.Order;
import com.cloud.rocketmqtxconsumer.entity.Pay;
import com.cloud.rocketmqtxconsumer.service.PayService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@SuppressWarnings("all")
@Component(value = "messageListener")
public class OrderMessageListener implements MessageListenerConcurrently {

    @Autowired
    private PayService payService;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        log.info("监听到事务消息，开始处理...");
        try{

            for (MessageExt msg : msgs) {
                if (payService.selectTransactionInformationById(msg.getTransactionId()) <= 0){
                    //转换消息
                    String data = new String(msg.getBody());
                    Order order = JSONObject.parseObject(data, Order.class);
                    //执行业务逻辑,消费消息
                    Pay pay = Pay.builder().payName(order.getOrderName()).build();
                    payService.insertPayInfo(pay,msg.getTransactionId());
                }else {
                    log.info("消息已经被消费过了");
                }
            }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }catch (Exception e){
            log.error("消费消息失败,稍后重试");
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
